package com.ssy.lingxi.marketing.handler.listener;

import cn.hutool.json.JSONUtil;
import com.rabbitmq.client.Channel;
import com.ssy.lingxi.component.redis.lock.LockException;
import com.ssy.lingxi.component.redis.lock.LockInfo;
import com.ssy.lingxi.component.redis.lock.LockTemplate;
import com.ssy.lingxi.marketing.api.model.constant.MkFeignConstants;
import com.ssy.lingxi.marketing.api.model.constant.MkQueueMessageTypeEnum;
import com.ssy.lingxi.marketing.api.model.constant.MkRedisKeyConstants;
import com.ssy.lingxi.marketing.api.model.dto.MarketingQueueDTO;
import com.ssy.lingxi.marketing.api.model.request.GiftCouponRequest;
import com.ssy.lingxi.marketing.api.model.request.GoodsSalesAddReq;
import com.ssy.lingxi.marketing.service.IActivityExecuteService;
import com.ssy.lingxi.marketing.service.IActivityOrderService;
import com.ssy.lingxi.marketing.service.IBaseMqService;
import com.ssy.lingxi.marketing.service.IMobileCouponService;
import com.ssy.lingxi.marketing.serviceimpl.component.coupon.CouponDetailExpireComponent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Objects;

/**
 *  营销服务-消息队列监听类
 * @author ds
 * @date 2022/3/8
 * @version 2.0.0
 */
@Component
public class MessageQueueListener {
    @Value("${spring.profiles.active:dev}")
    private String env;
    @Resource
    private IActivityExecuteService activityExecuteService;
    @Resource
    private IActivityOrderService activityOrderService;
    @Resource
    private CouponDetailExpireComponent couponDetailExpireComponent;
    @Resource
    private IMobileCouponService mobileCouponService;
    @Resource
    private IBaseMqService baseMqService;

    @Autowired
    private LockTemplate lockTemplate;


    private final Logger log = LoggerFactory.getLogger(this.getClass());

    /**
     * 迟队列消费者
     * @param message 消息
     * @param channel  消息通道
     */
    @RabbitListener(queues = MkFeignConstants.MK_DELAY_QUEUE, ackMode = "MANUAL")
    public void delayQueue(Message message, Channel channel) {
        String jsonMessage = new String(message.getBody());
        log.info("内部延迟队列接收到消息 => " + jsonMessage);
        try {
            MarketingQueueDTO queueDTO = JSONUtil.toBean(jsonMessage, MarketingQueueDTO.class);
            if(Objects.isNull(queueDTO) || Objects.isNull(queueDTO.getType()) || !StringUtils.hasLength(queueDTO.getMessage())) {
                log.info("内部延迟队列消息反序列化后错误 => " + jsonMessage);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
                return;
            }

            //根据延迟时间判断是否需要重新加入队列
            if(baseMqService.reScheduleDelayMessage(queueDTO)) {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
                return;
            }

            MkQueueMessageTypeEnum messageType = MkQueueMessageTypeEnum.parse(queueDTO.getType());
            switch (messageType) {
                case GROUP_PURCHASE_TIMEOUT:
                    activityOrderService.updateGroupPurchaseStatus(Long.parseLong(queueDTO.getMessage()));
                    break;
                case COUPON_EXPIRE:
                    couponDetailExpireComponent.handleMqMsgCouponDetailExpire(queueDTO.getMessage());
                    break;
                default:
                    break;
            }
            //确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        } catch (Exception e){
            log.error("内部延迟队列接收消息错误：jsonMessage:{}",jsonMessage ,e);
        }
    }

    /**
     * 普通列消费者
     * @param message 消息
     * @param channel  消息通道
     */
    @RabbitListener(queues = MkFeignConstants.MK_NORMAL_QUEUE, ackMode = "MANUAL")
    public void normalQueue(Message message, Channel channel) {
        String jsonMessage = new String(message.getBody());
        log.info("内部普通队列接收到消息 => " + jsonMessage);
        try {
            MarketingQueueDTO queueDTO = JSONUtil.toBean(jsonMessage, MarketingQueueDTO.class);
            if(Objects.isNull(queueDTO) || Objects.isNull(queueDTO.getType()) || !StringUtils.hasLength(queueDTO.getMessage())) {
                log.info("内部普通队列消息反序列化后错误 => " + jsonMessage);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
                return;
            }

            MkQueueMessageTypeEnum messageType = MkQueueMessageTypeEnum.parse(queueDTO.getType());
            switch (messageType) {
                case GIFT_COUPON:
                    mobileCouponService.handleMqMsgReceiveCoupon(JSONUtil.toBean(queueDTO.getMessage(), GiftCouponRequest.class));
                    break;
                default:
                    break;
            }
            //确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        }catch (Exception e){
            log.error("内部普通队列接收消息错误：jsonMessage:{}",jsonMessage ,e);
        }
    }

    /**
     * 更新活动商品销量-redis锁
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = MkFeignConstants.MK_GOODS_SALES_QUEUE,ackMode ="MANUAL")
    public void goodsSalesUpdate(Message message, Channel channel) throws IOException {
        String bodyStr = new String(message.getBody());
        LockInfo lock=null;
        String addKey=null;
        try {
            if(env.startsWith("dev")) {
                log.info("活动商品销量监听,收到消息：{}",bodyStr);
            }
            GoodsSalesAddReq req = JSONUtil.toBean(bodyStr, GoodsSalesAddReq.class);
            //加锁
            addKey = MkRedisKeyConstants.getActivityGoodsSalesAddKey(req.getBelongType(),req.getActivityId(),req.getSkuId());
            lock = lockTemplate.lock(addKey);
            if(lock!=null) {
                activityExecuteService.updateActivityGoodsSales(req);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }else{
                channel.basicRecover(true);
            }
        }catch (LockException le){
            log.error("处理退货处理队通知加锁异常,body:{},异常消息：",bodyStr,le);
        }catch (Exception e){
            log.error("处理退货处理队通知异常,body:{},异常消息：",bodyStr,e);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }finally {
            boolean result = lockTemplate.releaseLock(lock);
            if(!result&&lock!=null){
                log.warn("活动商品销量累计解锁失败：bodyStr:{},addKey:{}",bodyStr,addKey);
            }
        }
    }

}
